package com.sinozo.data.app.function;

import com.sinozo.data.bean.UserInstallBean;
import com.sinozo.data.common.DataConfig;
import org.apache.flink.streaming.api.functions.AssignerWithPeriodicWatermarks;
import org.apache.flink.streaming.api.watermark.Watermark;


/**
 * @author zhaoyb
 * @version 1.0
 * @description
 * @date 2023/5/16 17:13:18
 */
public class UserInstallTimestampsAndWatermarks implements AssignerWithPeriodicWatermarks<UserInstallBean> {


    //1小时,单位：毫秒
    private static final long MAX_OUT_OF_ORDERLINESS =  DataConfig.MAX_OUT_OF_ORDERLINESS;

    private long currentMaxTimestamp;


    @Override
    public Watermark getCurrentWatermark() {

        return new Watermark(currentMaxTimestamp - MAX_OUT_OF_ORDERLINESS);
    }

    @Override
    public long extractTimestamp(UserInstallBean element, long recordTimestamp) {
        long timestamp = element.getTs();
        currentMaxTimestamp = Math.max(timestamp, currentMaxTimestamp);
        return timestamp;
    }
}
